package com.github.common.extend.demo.config.transaction.distributed.message.adaptive;

import com.baomidou.mybatisplus.extension.conditions.query.LambdaQueryChainWrapper;
import com.baomidou.mybatisplus.extension.conditions.update.LambdaUpdateChainWrapper;
import com.github.common.extend.demo.config.transaction.distributed.message.XMessage;
import com.github.common.extend.demo.config.transaction.distributed.message.XMessageStorage;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.Iterator;

/**
 * 通过mybatis存储rocketmq消息
 * @author X1993
 * @date 2022/9/22
 * @description
 */
@Component
public class RocketXMessageStorage implements XMessageStorage<RocketMessageDTO> {

    @Autowired
    private RocketMessageMapper rocketMessageMapper;

    @Override
    public void save(RocketMessageDTO message) {
        RocketMessageDO rocketMessageDO = convert(message);
        LocalDateTime now = LocalDateTime.now();

        rocketMessageDO.setSendCount(0);
        rocketMessageDO.setStatus(XMessage.UNSENT);
        rocketMessageDO.setCreateTime(now);
        rocketMessageDO.setUpdateTime(now);

        rocketMessageMapper.insert(rocketMessageDO);
    }

    @Override
    public void update(RocketMessageDTO message) {
        RocketMessageDO rocketMessageDO = convert(message);
        rocketMessageDO.setUpdateTime(LocalDateTime.now());

        new LambdaUpdateChainWrapper<>(rocketMessageMapper)
                .eq(RocketMessageDO::getId ,rocketMessageDO.getId())
                .in(RocketMessageDO::getStatus ,XMessage.UNSENT ,XMessage.FAIL_RETRY)
                .update(rocketMessageDO);
    }

    @Override
    public Iterator<RocketMessageDTO> unsentMessages()
    {
        return new Iterator<RocketMessageDTO>(){

            private Iterator<RocketMessageDO> iterator;

            private Long maxId;

            @Override
            public boolean hasNext()
            {
                if (iterator != null && iterator.hasNext()) {
                    return true;
                }

                iterator = new LambdaQueryChainWrapper<>(rocketMessageMapper)
                        .in(RocketMessageDO::getStatus, XMessage.UNSENT, XMessage.FAIL_RETRY)
                        .gt(maxId != null ,RocketMessageDO::getId ,maxId)
                        .orderByAsc(RocketMessageDO::getId)
                        .last("LIMIT 100 ")
                        .list()
                        .iterator();

                return iterator.hasNext();
            }

            @Override
            public RocketMessageDTO next() {
                RocketMessageDO rocketMessageDO = iterator.next();
                maxId = rocketMessageDO.getId();
                return convert(rocketMessageDO);
            }
        };
    }

    private static RocketMessageDO convert(RocketMessageDTO rocketMessageDTO)
    {
        RocketMessageDO rocketMessageDO = new RocketMessageDO();
        BeanUtils.copyProperties(rocketMessageDTO ,rocketMessageDO);
        return rocketMessageDO;
    }

    private static RocketMessageDTO convert(RocketMessageDO rocketMessageDO)
    {
        RocketMessageDTO rocketMessageDTO = new RocketMessageDTO();
        BeanUtils.copyProperties(rocketMessageDO ,rocketMessageDTO);
        return rocketMessageDTO;
    }

}
